package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Operator;
import co.elastic.clients.elasticsearch._types.query_dsl.QueryBuilders;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.base.BaseEntity;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.query.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.google.common.collect.Lists;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.LocalDateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Resource
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private ExecutorService executorService;
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Resource
    private  CategoryFeignClient categoryFeignClient;
    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public List<Map<String, Object>> channel(Long category1Id) throws IOException {
        List<Map<String, Object>> maps = new ArrayList<>();
        Result<List<BaseCategory3>> topBaseCategory3 = categoryFeignClient.findTopBaseCategory3(category1Id);
        if (topBaseCategory3==null){
            return maps;
        }
        List<BaseCategory3> baseCategory3s = topBaseCategory3.getData();
        if (CollectionUtils.isEmpty(baseCategory3s) ){
            return maps;
        }
        //从三级分类中 获取三级分类id集合
        List<FieldValue>  category3Ids = baseCategory3s.stream().map(baseCategory3 -> FieldValue.of(baseCategory3.getId())).collect(Collectors.toList());


        //构建dsl获取搜索解析结果集
        SearchRequest searchRequest = SearchRequest.of(s-> s.index("albuminfo")
                .query(q-> q.terms(t->t.field("category3Id").terms(ts->ts.value(category3Ids))))
                .size(0)
                .aggregations("categoryAgg",agg-> agg.terms(t->t.field("category3Id").size(6))
                        .aggregations("hotscoreTop",sagg->sagg.topHits(t->t.size(6).sort(st->st.field(f->f.field("hotScore")))))));


        SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

        //从es取出结果集
        Map<String, Aggregate> aggregations = response.aggregations();
        if (CollectionUtils.isEmpty(aggregations)){
            return maps;
        }
        Aggregate aggregate = aggregations.get("categoryAgg");
        if (aggregate==null){
            return maps;
        }
        //获取三级分类聚合中的桶集合

        Map<Long, BaseCategory3> category3Map = baseCategory3s.stream()
                .collect(Collectors.toMap(BaseEntity::getId, baseCategory3 -> baseCategory3));

        List<LongTermsBucket> buckets = aggregate.lterms().buckets().array();
        buckets.forEach(bucket ->{
            //把桶转化为map结构
            Map<String,Object> map =new HashMap<>();
            maps.add(map);

            //向map中添加三级分类对象
            map.put("baseCategory3",category3Map.get(bucket.key()));

            //向map中添加 三级分类板块对应的热门数据
            //获取桶中的子聚合

            Aggregate aggregate1 = bucket.aggregations().get("hotscoreTop");

            List<Hit<JsonData>> hits = aggregate1.topHits().hits().hits();

            List<AlbumInfoIndex> collect = hits.stream().map(hit -> hit.source().to(AlbumInfoIndex.class)).collect(Collectors.toList());

            map.put("list",collect);
        });

        return maps;
    }



    @Override
    public AlbumSearchResponseVo searchAlbumInfo(AlbumIndexQuery albumIndexQuery) {
    try {


        SearchResponse<AlbumInfoIndex> response = this.elasticsearchClient.search(this.buildDsl(albumIndexQuery), AlbumInfoIndex.class);

        AlbumSearchResponseVo albumSearchResponseVo = this.parseResult(response,albumIndexQuery.getPageNo(),albumIndexQuery.getPageSize());

        return albumSearchResponseVo;
    }catch (Exception e){
        throw new RuntimeException(e.getMessage());
    }

    }




    @Override
    public List<String> completeSuggest(String keyword) {
        List<String> suggesters = new ArrayList<>();
        try {

            SearchRequest request = SearchRequest.of(s -> s.index("suggestinfo")
                    .suggest(sg->sg.suggesters("titleSuggest",ts->ts
                            .prefix(keyword)
                            .completion(c -> c.field("keyword").size(10).skipDuplicates(true)))
                    .suggesters("titlePySuggest",ts->ts
                            .prefix(keyword)
                            .completion(c -> c.field("keywordPinyin").size(10).skipDuplicates(true)))
                            .suggesters("titleSeqSuggest",ts->ts
                            .prefix(keyword)
                            .completion(c -> c.field("keywordSequence").size(10).skipDuplicates(true)))))
                    ;


            SearchResponse<SuggestIndex> response = this.elasticsearchClient.search(request, SuggestIndex.class);
            //获取提示词
            Map<String, List<Suggestion<SuggestIndex>>> suggest = response.suggest();
            //获取标题的提示词
            if (suggest==null){
                return null;
            }
            List<CompletionSuggestOption<SuggestIndex>> titleOptions = suggest.get("titleSuggest").get(0).completion().options();

            List<CompletionSuggestOption<SuggestIndex>> titlePySuggest = suggest.get("titlePySuggest").get(0).completion().options();

            List<CompletionSuggestOption<SuggestIndex>> titleSeqOptions = suggest.get("titleSeqSuggest").get(0).completion().options();

            if (!CollectionUtils.isEmpty(titleOptions)){
                titleOptions.forEach(ops ->{
                    suggesters.add(ops.source().getTitle());
                });
            }

            if (!CollectionUtils.isEmpty(titlePySuggest)){
                titlePySuggest.forEach(ops ->{
                    suggesters.add(ops.source().getTitle());
                });
            }
            if (!CollectionUtils.isEmpty(titleSeqOptions)){
                titleSeqOptions.forEach(ops ->{
                    suggesters.add(ops.source().getTitle());
                });
            }

            //进行联想搜索
            SearchResponse<SuggestIndex> suggestinfo = this.elasticsearchClient.search(s -> s.index("suggestinfo")
                    .query(q -> q.match(m -> m.field("title").query(keyword))), SuggestIndex.class);
            List<Hit<SuggestIndex>> hits = suggestinfo.hits().hits();
            hits.forEach(hit->{
                if (suggesters.size()<10){
                    suggesters.add(hit.source().getTitle());
                }
            });
            Set<String> titleSet =new HashSet<>();
            titleSet.addAll(suggesters);
            suggesters.clear();
            suggesters.addAll(titleSet);


            return suggesters;
        }catch (Exception e){
            throw new  RuntimeException();
        }
    }




    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String statType) {


        try {
            SearchResponse<AlbumInfoIndex> response = this.elasticsearchClient.search(s -> s.index("albuminfo")
                    .query(q -> q.term(t -> t.field("category1Id").value(category1Id)))
                    .sort(st -> st.field(f -> f.field(statType)))
                    .size(20), AlbumInfoIndex.class);

            List<Hit<AlbumInfoIndex>> hits = response.hits().hits();
            List<AlbumInfoIndexVo> albumInfoIndexVos = hits.stream().map(hit -> {
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                BeanUtils.copyProperties(hit.source(), albumInfoIndexVo);
                return albumInfoIndexVo;
            }).collect(Collectors.toList());


            return albumInfoIndexVos;

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }




    @Override
    public void syncDataToEs() {
        //获取状态发生变化的专辑Id集合
        //获取上一次任务的截止时间
        String  startTime = (String) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_STAT_ENDTIME);

        if (StringUtils.isBlank(startTime)){
            startTime="2023-06-05 22:30:00"; //取一个默认的时间作为起始时间
        }



        //取系统当前时间作为截止时间，并放入到redis中
        String endTime = LocalDateTime.now().toString("yyyy-MM-dd HH:mm:ss");
        Result<List<Long>> changedAlbumIds = this.albumInfoFeignClient.getChangedAlbumIds(startTime, endTime);
        Assert.notNull(changedAlbumIds,"更新ES中的专辑状态，远程调用获取状态变化的专辑ids失败");
        List<Long> albumIdsData = changedAlbumIds.getData();
        if (albumIdsData==null){
            return;
        }


        //根据专辑id集合获取变化后的状态
        //考虑到获取到的状态变化的专辑id数据量庞大 先分成一小组 之后使用异步编排
        //使用CountDownLatch
        List<List<Long>> albumIdspartition = Lists.partition(albumIdsData, 1000);
        CountDownLatch countDownLatch = new CountDownLatch(albumIdspartition.size());
        albumIdspartition.forEach(albumSubIds ->{
            executorService.execute(()->{
                Result<List<AlbumStatVo>> albumStatsByAlbumIds = this.albumInfoFeignClient.getAlbumStatsByAlbumIds(albumSubIds);
                Assert.notNull(albumStatsByAlbumIds,"更新专辑状态到es中，远程调用获取专辑变化后的数据失败");
                List<AlbumStatVo> albumStatVos = albumStatsByAlbumIds.getData();
                albumStatVos.forEach(albumStatVo -> {
                    //同步到ES中
                    AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
                    albumInfoIndex.setId(albumStatVo.getAlbumId());
                    albumInfoIndex.setHotScore(albumStatVo.getPlayStatNum()*0.1+albumStatVo.getBuyStatNum()*0.4
                                                +albumStatVo.getCommentStatNum()*0.2+albumStatVo.getSubscribeStatNum()*0.3);
                    BeanUtils.copyProperties(albumStatVo,albumInfoIndex);
                    this.elasticsearchTemplate.update(albumInfoIndex);
                });
                countDownLatch.countDown();

            });


        });
        try {
            countDownLatch.await();
        }catch (Exception e){
            throw new RuntimeException(e);
        }


    }






    //构建DSL
    private SearchRequest buildDsl(AlbumIndexQuery query){
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index("albuminfo");
        //1.查询过滤
        BoolQuery.Builder boolQueryBuilder = QueryBuilders.bool();
        String keyword = query.getKeyword();

        if (StringUtils.isNotBlank(keyword)){
            boolQueryBuilder.must(m->m.multiMatch(mm->mm.query(query.getKeyword())
                    .fields("albumTitle",
                            "announcerName",
                            "albumIntro")
                    .query(keyword)
                    .operator(Operator.And)));

        }
        //1.2构建分类过滤
        Long category1Id = query.getCategory1Id();
        if (category1Id!=null) {
            boolQueryBuilder.filter(f->f.term(t->t.field("category1Id").value(category1Id)));
        }
        Long category2Id = query.getCategory2Id();
        if (category2Id!=null) {
            boolQueryBuilder.filter(f->f.term(t->t.field("category2Id").value(category2Id)));
        }
        Long category3Id = query.getCategory3Id();
        if (category3Id!=null) {
            boolQueryBuilder.filter(f->f.term(t->t.field("category3Id").value(category3Id)));
        }

        //1.3 构建 标签的嵌套过滤
        List<String> attributeList = query.getAttributeList();
        if (!CollectionUtils.isEmpty(attributeList)){
            //遍历标签  1:1  2:3
            attributeList.forEach(attribute->{ //1:1
             String[] attrAndValue=   StringUtils.split(attribute,":");
             //校验当前字符串是否合法，长度必须是2并且都必须是纯数字 使用StringUtils(lang3).isNumeric来判断
             if (attrAndValue==null ||attrAndValue.length!=2 ||!StringUtils.isNumeric(attrAndValue[0]) || !StringUtils.isNumeric(attrAndValue[1])){
                 return;
             }
             //添加嵌套过滤
                boolQueryBuilder.filter(f->f.nested(n->n
                        .path("attributeValueIndexList")
                        .query(q    ->  q.bool( b-> b
                                .must(m1->m1.term(t->t.field("attributeValueIndexList.attributeId").value(attrAndValue[0])))
                                .must(m2->m2.term(t->t.field("attributeValueIndexList.valueId").value(attrAndValue[1])))))));

            });

        }

        //必须先放条件 最后build()
        builder.query(boolQueryBuilder.build()._toQuery());
        //2.排序
        String order = query.getOrder();
        if (!StringUtils.isEmpty(order)){
            String[] sort = StringUtils.split(order, ":");
            if (sort!=null && sort.length==2 && StringUtils.isNumeric(sort[0]) && List.of("desc","asc").contains(sort[1])){
                String field = "";
                switch (sort[0]){
                    case "1": field ="hotScore"; break;
                    case "2": field = "playStatNum"; break;
                    case "3": field = "createTime"; break;
                    default: field="_score";
                }
        String finalField = field;

                builder.sort(s->s.field(f->f.field(finalField)
                        .order(StringUtils.equals(sort[1],"asc")? SortOrder.Asc:SortOrder.Desc))
                );
            }
        }
        //3.分页

        Integer pageNo = query.getPageNo();
        Integer pageSize = query.getPageSize();

        builder.from((pageNo-1)*pageSize);
        builder.size(pageSize);


        //4.高亮
        if (StringUtils.isNotEmpty(keyword)){

            builder.highlight(h-> h.fields("albumTitle",f->f.preTags("<font style='color:red;'>").postTags("</font>"))
                    .fields("albumIntro",f->f.preTags("<font style='color:red;'>").postTags("</font>")));
        }


        //5.结果集过滤
        builder.source(sc->sc.filter(f->f.excludes("category1Id","category2Id","category3Id","hotScore","attributeValueIndexList")));
        SearchRequest request = builder.build();

        System.out.println(request);
        return request;
    }


    //解析结果集
    private AlbumSearchResponseVo parseResult(SearchResponse<AlbumInfoIndex> response,Integer pageNo,Integer pageSize){
        AlbumSearchResponseVo responseVo = new AlbumSearchResponseVo();
        //外层hits
        HitsMetadata<AlbumInfoIndex> hits = response.hits();

        responseVo.setTotal(hits.total().value());
        long totalPage = responseVo.getTotal()/pageSize;
        responseVo.setTotalPages(responseVo.getTotal()%pageSize==0?totalPage:totalPage+1);
        //分页参数从请求参数中获取
        responseVo.setPageNo(pageNo);
        responseVo.setPageSize(pageSize);

        //当前页的数据
        List<Hit<AlbumInfoIndex>> hitList = hits.hits();



        responseVo.setList(hitList.stream().map(hit->{
            AlbumInfoIndex albumInfoIndex = hit.source();
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex,albumInfoIndexVo);
            //解析高亮结果集
            Map<String, List<String>> highlight = hit.highlight();
            if (!CollectionUtils.isEmpty(highlight)){
                List<String> albumTitles = highlight.get("albumTitle");
                if (!CollectionUtils.isEmpty(albumTitles)){
                    albumInfoIndexVo.setAlbumTitle(albumTitles.get(0));
                }
                //简介的高亮结果集
                List<String> albumIntro = highlight.get("albumIntro");
                if (!CollectionUtils.isEmpty(albumIntro)){
                    albumInfoIndexVo.setAlbumIntro(albumIntro.get(0));
                }
            }

            return albumInfoIndexVo;
        }).collect(Collectors.toList()));
        return responseVo;
    }


}
